//********************************************************************
//	created:	13:10:2011   10:41
//	filename: 	server.cpp
//	author:		tiamo
//	purpose:	service
//********************************************************************

#include "stdafx.h"
#include "server.h"
#include "rssservice.h"
#include "apnservice.h"
#include "rpcservice.h"

namespace sql
{
	//
	// build notification
	//
	class BuildNotification : public mysql::SQLRequest
	{
	public:
		//
		// constructor
		//
		BuildNotification(mysql::Database& db) : SQLRequest(db, L"BuildNotification") {}

	private:
		//
		// do request imp
		//
		virtual void doRequestImp();

		//
		// build payload
		//
		UniString buildPayload(UniString const& prefix, UniString rawTitle, UniString suffix);

	public:
		//
		// content table name
		//
		UniString															mContentTableName;

		//
		// rss name
		//
		UniString															mServiceName;

		//
		// notification count
		//
		uint32_t															mItemCount;
	};
}

//
// constructor
//
Server::Server()
	: mHttpRequestManager(mStopEvent), mSQLRequestManager(mStopEvent)
{
	mSocketManager															= nullptr;
	mSingleton																= this;
}

//
// destructor
//
Server::~Server()
{
	delete mSocketManager;
	std::for_each(mServiceList.begin(), mServiceList.end(), [](Service* theService){delete theService;});
	mSingleton																= nullptr;
}

//
// start
//
void Server::start(int argc, char const* argv[])
{
	//
	// initialize openssl
	//
	SSL_library_init();

	//
	// initialize curl
	//
	curl_global_init(CURL_GLOBAL_DEFAULT);

	//
	// load global config
	//
#ifdef DEBUG
	mConfig.openConfigFile(argc > 1 ? utils::utf8ToWide(argv[1]).c_str() : L"debug.cfg");
#else
	mConfig.openConfigFile(argc > 1 ? utils::utf8ToWide(argv[1]).c_str() : L"release.cfg");
#endif

	//
	// initialize log manager
	//
	mLogManager.initialize(mConfig.getString(L"/config/log/path", L"log/u2daemon").c_str(), mConfig.getInt(L"/config/log/level", utils::LogManager::LevelWarning), mConfig.getBoolean(L"/config/log/syslog", false));

	//
	// open database
	//
	mDatabase.open(mConfig.getString(L"/config/db/host"), mConfig.getString(L"/config/db/user"), mConfig.getString(L"/config/db/pass"), mConfig.getString(L"/config/db/db"));

	//
	// start http/sql manager
	//
	mHttpRequestManager.start();
	mSQLRequestManager.start();

	//
	// determinate socket manager
	//
#ifndef WIN32
	UniString socketManagerType												= mConfig.getString(L"/config/socket_manager/type", L"epoll");
	if(socketManagerType == L"epoll")
		mSocketManager														= new network::EpollSocketManager(mStopEvent);
#endif

	//
	// default is select
	//
	if(!mSocketManager)
		mSocketManager														= new network::SelectSocketManager(mStopEvent);

	//
	// start socket manager
	//
	mSocketManager->start(&mConfig);

	//
	// create browser socket
	//
	if(mConfig.getBoolean(L"/config/rpc/enable", true))
		startService(new rpc::Service(L"scgi stub", mDatabase));

	//
	// start apn service
	//
	if(mConfig.getBoolean(L"/config/apn/enable", true))
	{
		startService(new apn::Service(L"APN provider (develpment)", mDatabase, true));
		startService(new apn::Service(L"APN provider (production)", mDatabase, false));
	}

	//
	// start rss service
	//
	if(mConfig.getBoolean(L"/config/rss/enable", true))
	{
		rss::Service* rssService											= new rss::Service(L"u2@dmhy", mDatabase);
		rssService->setNewItemCallback(std::bind(&Server::onNewRSSItem, this, std::placeholders::_1, std::placeholders::_2));
		startService(rssService);
	}
}

//
// run
//
void Server::run()
{
	//
	// run as timer thread
	//
	while(!mStopEvent.wait(30))
	{
		mTimerManager.runFrame();
		mHttpRequestManager.processFinishedRequest();
		mSQLRequestManager.processFinishedRequest();
		mSocketManager->processSocketEvent();
	}

	//
	// cleanup
	//
	std::for_each(mServiceList.begin(), mServiceList.end(), [](Service* theService){theService->cleanup();});

	//
	// wait threads
	//
	LogInfo(L"Service(%p): stop event detected, wait http/sql/socket threads finish", this);
	mHttpRequestManager.wait();
	mSQLRequestManager.wait();
	mSocketManager->wait();
	LogInfo(L"Service(%p): all threads finished, done", this);
}

//
// stop
//
void Server::stop()
{
	LogInfo(L"Service(%p): signal stop event", this);
	mStopEvent.signal();
}

//
// start service
//
void Server::startService(Service* theService)
{
	LogInfo(L"Server(%p): starting service(%p:%ls)", this, theService, theService->getName().c_str());
	mServiceList.push_back(theService);
	theService->initialize();
}

//
// get service by name
//
Service* Server::getServiceByName(wchar_t const* serviceName)
{
	for(auto it = mServiceList.begin(); it != mServiceList.end(); ++ it)
	{
		Service* theService													= *it;
		if(theService->getName() == serviceName)
			return theService;
	}

	return nullptr;
}

//
// new rss item
//
void Server::onNewRSSItem(rss::Service* theService, rss::RSSItemList const& addedItemList)
{
	//
	// build notification
	//
	utils::SmartPointer<sql::BuildNotification> buildNotification			= new sql::BuildNotification(mDatabase);
	buildNotification->mContentTableName									= theService->getTableName();
	buildNotification->mServiceName											= theService->getName();
	buildNotification->setCallback(std::bind(&Server::onBuildNotification, this, std::placeholders::_1));
	LogInfo(L"Server(%p): building notification(%u) for (%ls)", this, static_cast<uint32_t>(addedItemList.size()), buildNotification->mServiceName.c_str());
	mSQLRequestManager.pushPendingRequest(buildNotification.get());
}

//
// build notification
//
void Server::onBuildNotification(utils::BaseRequest* theRequest)
{
	//
	// check result
	//
	sql::BuildNotification* buildNotification								= static_cast<sql::BuildNotification*>(theRequest);
	LogInfo(L"Server(%p): build notification(%u) for (%ls) finished(%d)", this, buildNotification->mItemCount, buildNotification->mServiceName.c_str(), theRequest->getResult());
}

//
// build payload
//
UniString sql::BuildNotification::buildPayload(UniString const& prefix, UniString rawTitle, UniString suffix)
{
	bool insertDot															= true;
	while(1)
	{
		UniString title														= utils::encodeJSON(rawTitle);
		UniString payload													= prefix + title + suffix;
		if(utils::wideToUtf8(payload).length() < 200)
			return payload;

		if(insertDot)
		{
			suffix.insert(0, L"...");
			insertDot														= false;
		}

		if(rawTitle.length() <= 5)
			break;

		rawTitle															= rawTitle.substr(0, rawTitle.length() - 5);
	}

	return L"";
}

//
// build notification
//
void sql::BuildNotification::doRequestImp()
{
	//
	// start transaction
	//
	mDatabase.startTransaction();
	auto rollback															= utils::makeResourceHelper([this](){mDatabase.rollbackTransaction();});

	//
	// load device list
	//
	struct device
	{
		UniString															mToken;
		UniString															mFilter;
		uint8_t																mDevMode;
		uint32_t															mMaxCount;
	}dbDevice;
	std::vector<device> deviceList;
	mysql::Statement statement												= mDatabase.createStatement();
	statement.prepare(L"select device_token, category_filter,dev_mode,max_count from subscription where unix_timestamp(delete_time)=0 and service_name=?");
	statement.bind(L"device_token", false, dbDevice.mToken, 64);
	statement.bind(L"category_filter", false, dbDevice.mFilter, 255);
	statement.bind(L"dev_mode", false, dbDevice.mDevMode);
	statement.bind(L"max_count", false, dbDevice.mMaxCount);
	statement.bind(L"service_name", true, mServiceName);
	statement.execute();
	while(statement.fetch())
		deviceList.push_back(dbDevice);

	//
	// load content list
	//
	struct content
	{
		UniString															mTitle;
		uint32_t															mCategoryId;
	}dbContent;
	std::vector<content> contentList;
	statement.prepareV(L"select title, category_id from %ls where `read`=0 order by publish desc", mContentTableName.c_str());
	statement.bind(L"title", false, dbContent.mTitle);
	statement.bind(L"category_id", false, dbContent.mCategoryId);
	statement.execute();
	while(statement.fetch())
		contentList.push_back(dbContent);

	//
	// filter item list
	//
	struct notification_item
	{
		device*																mDevice;
		content*															mContent;
		notification_item(device* dbDevice = nullptr, content* dbContent = nullptr) : mDevice(dbDevice), mContent(dbContent) {}
	};
	std::map<UniString, std::list<notification_item>> itemList;
	for(auto itDevice = deviceList.begin(); itDevice != deviceList.end(); ++ itDevice)
	{
		device& dbDevice													= *itDevice;
		for(auto itContent = contentList.begin(); itContent != contentList.end(); ++ itContent)
		{
			content& dbContent												= *itContent;
			wchar_t category[64];
			platform::snwprintf(category, ARRAYSIZE(category), L",%d,", dbContent.mCategoryId);
			if(wcsstr(dbDevice.mFilter.c_str(), category))
				continue;

			std::list<notification_item>& temp								= itemList[dbDevice.mToken];
			notification_item item(&dbDevice, &dbContent);
			temp.push_back(item);

			if(temp.size() < dbDevice.mMaxCount)
				continue;

			temp.resize(1);
			dbDevice.mMaxCount												= 0;
			break;
		}
	}

	//
	// build payload
	//
	mItemCount																= 0;
	statement.prepareV(L"insert into notification set device_token=?, payload=?, dev_mode=?, expire=from_unixtime(%u)", static_cast<uint32_t>(time(0) + 24 * 60 * 60));
	for(auto it = itemList.begin(); it != itemList.end(); ++ it)
	{
		for(auto it2 = it->second.begin(); it2 != it->second.end(); ++ it2)
		{
			notification_item& item											= *it2;
			device& dbDevice												= *item.mDevice;
			content& dbContent												= *item.mContent;
			wchar_t suffix[255]												= {0};
			uint32_t badgeCount												= static_cast<uint32_t>(contentList.size());
			platform::snwprintf(suffix, ARRAYSIZE(suffix), L"\",\"badge\":%u,\"sound\":\"default\"},\"svc\":\"%ls\"}", badgeCount, mServiceName.c_str());
			wchar_t prefix[255]												= L"{\"aps\":{\"alert\":\"";
			if(!dbDevice.mMaxCount && badgeCount > 1)
				platform::snwprintf(prefix, ARRAYSIZE(prefix), L"{\"aps\":{\"alert\":\"(+%u) ", static_cast<uint32_t>(contentList.size()));

			UniString payload												= buildPayload(prefix, dbContent.mTitle, suffix);
			if(payload.empty())
				continue;

			mItemCount														+= 1;
			statement.reset(false, true);
			statement.bind(L"device_token", true, const_cast<UniString&>(dbDevice.mToken));
			statement.bind(L"payload", true, payload);
			statement.bind(L"dev_mode", true, dbDevice.mDevMode);
			statement.execute();
		}
	}

	//
	// set read flag
	//
	statement.prepareV(L"update %ls set `read`=1 where `read`=0", mContentTableName.c_str());
	statement.execute();

	mDatabase.commitTransaction();
	rollback.dismiss();
}

//
// singleton
//
Server* Server::mSingleton													= nullptr;
